package com.sojson.project.sys.sharding.service.impl;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import javax.annotation.Resource;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.sojson.config.datasource.annotation.DataSource;
import com.sojson.config.datasource.enums.EDataSourceType;
import com.sojson.project.sys.sharding.dao.DataTransferDao;
import com.sojson.project.sys.sharding.entity.dto.DataTransferDto;
import com.sojson.project.sys.sharding.entity.po.DataTransfer;
import com.sojson.project.sys.sharding.service.IDataTransferService;
import com.sojson.util.DateUtil;
import com.sojson.util.StringUtil;
import com.sojson.util.db.DbUtil;
import com.sojson.util.db.ShardingJdbcUtil;
import com.sojson.util.db.bean.ShardingJdbcVo;

import lombok.extern.slf4j.Slf4j;

/**
 * 数据迁移工具类
 *
 * @author liu
 * @date 2024-04-25
 */
@Slf4j
@Component
public class DataTransferServiceImpl extends ServiceImpl<DataTransferDao, DataTransfer>
    implements IDataTransferService {

    @Lazy
    @Autowired
    private DataTransferServiceImpl dataTransferUtil;
    @Resource
    private DataTransferDao dataTransferDao;
    /** 单次查询量 */
    public int searchSize = 200000;
    /** 单次处理量 */
    public int disposeSize = 10000;

    @Override
    @DataSource(EDataSourceType.master)
    public void execute() {
        log.info("开始执行数据归档: " + LocalDateTime.now());
        long start = System.currentTimeMillis();

        LambdaQueryWrapper<DataTransfer> query = Wrappers.lambdaQuery(DataTransfer.class);
        query.eq(DataTransfer::getStatus, true);
        List<DataTransfer> tables = dataTransferDao.selectList(query);
        if (tables == null || tables.size() < 1) {
            return;
        }
        CountDownLatch latch = new CountDownLatch(tables.size());
        try {
            for (DataTransfer table : tables) {
                try {
                    dataTransferUtil.transfer(table, latch);
                } catch (Exception e) {
                    latch.countDown();
                    log.error(e.getMessage(), e);
                }
            }
            latch.await();
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }

        log.info("数据归档执行结束，" + "共用时: " + (System.currentTimeMillis() - start));
    }

    @Async
    public void transfer(DataTransfer table, CountDownLatch latch) {
        try {
            table = dataTransferDao.selectById(table.getTableName());
            long start = System.currentTimeMillis();
            log.info(table.getTableName() + "开始执行数据归档: " + LocalDateTime.now());

            run(table);

            log.info(table.getTableName() + "数据归档共用时: " + (System.currentTimeMillis() - start) / 1000 + "秒");
        } finally {
            latch.countDown();
        }
    }

    public void run(DataTransfer dto) {
        dto = dataTransferDao.selectById(dto.getTableName());

        String prefix = "#{data.";
        String suffix = "},";
        // 获取历史数据分组信息
        DataTransferDto dataTransfer = getDataTransfer(dto);
        try {
            // 创建表
            createTable(dataTransfer);

            // 查询要迁移的字段
            String fields = dataTransferDao.getFields(dataTransfer).toUpperCase();
            dataTransfer.setFieldStrs(fields);
            StringBuilder builder = new StringBuilder();
            String[] split = fields.split(",");
            for (String field : split) {
                builder.append(prefix);
                builder.append(field);
                builder.append(suffix);
            }
            dataTransfer.setFieldsqls(builder.substring(0, builder.length() - 1));

            // 迁移数据
            dataTransferUtil.transfer(dataTransfer);

            // 记录归档日期
            boolean updateLastDate = dataTransferUtil.updateLastDate(dataTransfer);
            if (updateLastDate) {
                run(dto);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 获取表信息
     *
     * @param dto
     * @return
     */
    public DataTransferDto getDataTransfer(DataTransfer dto) {
        String databaseName = dataTransferDao.getDatabaseName();
        String tableName = dto.getTableName();
        String fieldNameId = dto.getFieldNameId();
        String fieldNameSearch = dto.getFieldNameSearch();
        Integer archiveInterval = dto.getArchiveInterval();
        // 当前时间
        LocalDate now = DateUtil.stringToLocalDate(DateUtil.dateToString(DateUtil.nowLocalDate(), DateUtil.YYYYMM01),
            DateUtil.YYYYMMDD);
        // 上一次归档结束时间
        LocalDate lastTime = dto.getLastTime();
        // 归档结束时间
        LocalDate endDateTransfer = lastTime.plusMonths(archiveInterval);
        // 数据结束时间
        LocalDate endDateDate = now.plusMonths(-archiveInterval + 1);
        // 结束时间
        LocalDate endDate = null;
        if (DateUtil.dateToLong(endDateDate) < DateUtil.dateToLong(endDateTransfer)) {
            endDate = endDateDate;
        } else {
            endDate = endDateTransfer;
        }
        // 表后缀
        String suffix = DateUtil.dateToString(lastTime, DateUtil.YYYY_MM);
        DataTransferDto dataTransfer = new DataTransferDto();
        dataTransfer.setDatabaseName(databaseName);
        dataTransfer.setTableName(tableName);
        dataTransfer.setTableNameBak(tableName + StringUtil.SEPARATOR + suffix);
        dataTransfer.setFieldNameSearch(fieldNameSearch.toUpperCase());
        dataTransfer.setFieldNameId(fieldNameId.toUpperCase());
        dataTransfer.setNowDate(DateUtil.localDateTimeToDate(now));
        dataTransfer.setStartDate(DateUtil.localDateTimeToDate(lastTime));
        dataTransfer.setEndDateTransfer(DateUtil.localDateTimeToDate(endDateTransfer));
        dataTransfer.setEndDateDate(DateUtil.localDateTimeToDate(endDateDate));
        dataTransfer.setEndDate(DateUtil.localDateTimeToDate(endDate));
        Integer searchSize = dto.getSearchSize();
        dataTransfer.setSearchSize(searchSize == null ? this.searchSize : searchSize);
        Integer disposeSize = dto.getDisposeSize();
        dataTransfer.setDisposeSize(disposeSize == null ? this.disposeSize : disposeSize);

        Map<String, ShardingJdbcVo> tables = ShardingJdbcUtil.getTables();
        dataTransfer.setSharding(tables.containsKey(tableName));
        return dataTransfer;
    }

    /**
     * 创建表
     *
     * @param dto
     */
    public void createTable(DataTransferDto dto) {
        String databaseName = dto.getDatabaseName();
        String tableName = dto.getTableName();
        String tableNameBak = dto.getTableNameBak();
        if (dto.isSharding()) {
            dto.setTableName(tableName + "_1");
            DbUtil.push(EDataSourceType.sharding);
        } else {
            DbUtil.push(EDataSourceType.master);
        }

        // 判断备份表是否存在
        int tableIsExists = dataTransferDao.tableIsExists(databaseName, tableNameBak);
        if (tableIsExists > 0) {
            return;
        }

        // 判断数据表是否存在
        tableIsExists = dataTransferDao.tableIsExists(databaseName, tableName);
        if (tableIsExists < 1) {
            throw new RuntimeException(tableName + "数据表不存在");
        }

        // 获取建表语句
        Map<String, String> createTableSql = dataTransferDao.getCreateTableSql(dto);
        if (createTableSql == null) {
            throw new RuntimeException("表不存在");
        }
        String sql = createTableSql.get("Create Table");
        if (StringUtils.isBlank(sql)) {
            throw new RuntimeException("建表语句获取失败");
        }
        sql = sql.replace(String.format("TABLE `%s`", tableName),
            String.format("TABLE IF NOT EXISTS `%s`", tableNameBak));

        // 创建表
        dataTransferDao.createTable(sql);

        dto.setTableName(tableName);
    }

    /**
     * 迁移数据
     *
     * @param dto
     */
    public void transfer(DataTransferDto dto) {
        // 查询数据
        List<Map<String, Object>> datas = dataTransferDao.getDatas(dto);
        if (datas.size() < 1) {
            return;
        }

        List<List<Map<String, Object>>> subLists = Lists.partition(datas, dto.getDisposeSize());
        for (List<Map<String, Object>> list : subLists) {
            // 保存数据
            dataTransferUtil.saveRun(dto, list);
        }

        // 重复执行，直到数据迁完
        dataTransferUtil.transfer(dto);
    }

    /**
     * 迁移数据
     *
     * @param dto
     */
    @Transactional(rollbackFor = Exception.class)
    public void saveRun(DataTransferDto dto, List<Map<String, Object>> datas) {
        // 保存数据
        int save = dataTransferDao.save(dto, datas);
        if (save != datas.size()) {
            throw new RuntimeException("数据保存失败,更新数量不匹配");
        }

        // 删除历史数据
        List<Object> ids = new ArrayList<>();
        for (Map<String, Object> map : datas) {
            ids.add(map.get(dto.getFieldNameId()));
        }

        int remove = dataTransferDao.remove(dto, ids);
        if (remove != datas.size()) {
            throw new RuntimeException("数据保存失败,更新数量不匹配");
        }
    }

    /**
     * 迁移数据
     *
     * @param dto
     */
    @Transactional(rollbackFor = Exception.class)
    public void saveOrUpdateRun(DataTransferDto dto, Map<String, Object> data) {
        // 保存数据
        int save = dataTransferDao.add(dto, data);
        if (save < 1) {
            throw new RuntimeException("数据保存失败");
        }

        dataTransferDao.removeById(dto, data.get(dto.getFieldNameId()));
    }

    /**
     * 记录归档日期
     *
     * @param dto
     */
    public boolean updateLastDate(DataTransferDto dto) {
        Date endDate = dto.getEndDate();
        Date endDateTransfer = dto.getEndDateTransfer();

        if (endDate.getTime() == endDateTransfer.getTime()) {
            DataTransfer transfer = new DataTransfer();
            transfer.setTableName(dto.getTableName());
            transfer.setLastTime(DateUtil.dateToLocalDate(endDate));
            dataTransferDao.updateById(transfer);
            return true;
        } else {
            return false;
        }
    }

}